package com.lazar.bigdata.hbase;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.io.IOException;

public class HbaseClient {
    Configuration conf = null;
    Connection conn = null;

    @Before
    public void init() {

        System.setProperty("hadoop.home.dir", "/Users/lazar/Downloads/hadoop-2.9.2");
        // 创建一个configuration对象
        conf = HBaseConfiguration.create();

       // conf.set("hbase.rootdir","hdfs://node51:9000/hbase");
        // 设置连接信息
        conf.set("hbase.zookeeper.quorum","node51");
        conf.set("hbase.zookeeper.property.clientPort","2181");
        // 创建连接
        try {
            conn = ConnectionFactory.createConnection(conf);
        } catch (IOException e) {
            e.printStackTrace();
            System.out.println("连接hbase失败");
        }
    }
    // 建表用户关系表
    @Test
    public void CreateUserMapperTable() throws IOException {
        // 获取hbase管理角色
        HBaseAdmin admin = (HBaseAdmin) conn.getAdmin();
        // 创建表描述器
        final HTableDescriptor des = new HTableDescriptor(TableName.valueOf("usermapper"));
        // 指定列族
        des.addFamily(new HColumnDescriptor("friends"));
        admin.createTable(des);
        System.out.println("用户关系表创建成功");
    }
    // 插入数据
    @Test
    public void putData() throws IOException {
        // 获取表对象
        Table t = conn.getTable(TableName.valueOf("usermapper"));
        // 获取put对象 指定rowkey
        Put put = null;

        put = new Put(Bytes.toBytes("uid1"));
        put.addColumn(Bytes.toBytes("friends"),Bytes.toBytes("uid2"),Bytes.toBytes("uid2"));
        put.addColumn(Bytes.toBytes("friends"),Bytes.toBytes("uid3"),Bytes.toBytes("uid3"));
        put.addColumn(Bytes.toBytes("friends"),Bytes.toBytes("uid5"),Bytes.toBytes("uid5"));
        // 插入数据
        t.put(put);
        put = new Put(Bytes.toBytes("uid2"));
        put.addColumn(Bytes.toBytes("friends"),Bytes.toBytes("uid1"),Bytes.toBytes("uid1"));
        put.addColumn(Bytes.toBytes("friends"),Bytes.toBytes("uid3"),Bytes.toBytes("uid3"));
        // 插入数据
        t.put(put);
        put = new Put(Bytes.toBytes("uid3"));
        put.addColumn(Bytes.toBytes("friends"),Bytes.toBytes("uid1"),Bytes.toBytes("uid1"));
        put.addColumn(Bytes.toBytes("friends"),Bytes.toBytes("uid2"),Bytes.toBytes("uid2"));
        put.addColumn(Bytes.toBytes("friends"),Bytes.toBytes("uid4"),Bytes.toBytes("uid4"));
        // 插入数据
        t.put(put);
        put = new Put(Bytes.toBytes("uid4"));
        put.addColumn(Bytes.toBytes("friends"),Bytes.toBytes("uid3"),Bytes.toBytes("uid3"));
        // 插入数据
        t.put(put);
        put = new Put(Bytes.toBytes("uid5"));
        put.addColumn(Bytes.toBytes("friends"),Bytes.toBytes("uid1"),Bytes.toBytes("uid1"));
        // 插入数据
        t.put(put);
        // 关闭table
        t.close();
        System.out.println("数据插入完毕");
    }

    // 查询数据
    //全表扫描
    @Test
    public void scanData() throws IOException {
        //准备table对象
        final Table table = conn.getTable(TableName.valueOf("usermapper"));
        //准备scan对象
        final Scan scan = new Scan();

        //执行扫描
        final ResultScanner resultScanner = table.getScanner(scan);
        for (Result result : resultScanner) {
            //获取到result中所有cell对象
            final Cell[] cells = result.rawCells();
            //遍历打印
            for (Cell cell : cells) {
                // 取出rowkey
                final String rowkey = Bytes.toString(CellUtil.cloneRow(cell));
                // 取出 列族
                final String f = Bytes.toString(CellUtil.cloneFamily(cell));
                // 取出列名
                final String column = Bytes.toString(CellUtil.cloneQualifier(cell));
                // 取出 数据值
                final String value = Bytes.toString(CellUtil.cloneValue(cell));
                System.out.println("rowkey-->" + rowkey + "--;cf-->" + f + ";column--->" + column + "--;value-->" + value);
            }
        }

        table.close();
    }

    //删除一条数据
    @Test
    public void deleteData() throws IOException {
        //需要获取一个table对象
        final Table table = conn.getTable(TableName.valueOf("usermapper"));

        //准备delete对象
        //删除用户1当好友5
        final Delete delete = new Delete(Bytes.toBytes("uid1"));
        delete.deleteColumns(Bytes.toBytes("friends"),Bytes.toBytes("uid5"));
        //执行删除
        table.delete(delete);

        //删除用户5的好友1
        final Delete delete2 = new Delete(Bytes.toBytes("uid5"));
        delete2.deleteColumns(Bytes.toBytes("friends"),Bytes.toBytes("uid1"));
        //执行删除
        table.delete(delete2);

        //关闭table对象
        table.close();
        System.out.println("删除数据成功！！");
    }

    //释放连接
    @After
    public void realse() {
        if (conn != null) {
            try {
                conn.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
}
